package com.message.sink;

import com.alibaba.fastjson.JSON;
import com.message.domain.MessageSourceDto;
import com.message.util.DingdingUtil;
import com.message.util.JedisPoolUtil;
import com.message.util.ThreadPoolUtil;
import com.message.util.UuidUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.MDC;
import redis.clients.jedis.Jedis;
import static com.message.consts.MessageConst.*;

@Slf4j
public class DingDingMessageSink extends RichSinkFunction<MessageSourceDto> {
    @Override
    public void invoke(MessageSourceDto value, Context context) {
        ThreadPoolUtil.getExecutorService().execute(() -> {
            String logId = UuidUtil.getLogId();//日志追踪号
            MDC.put("UUID", logId);
            Jedis jedisObject = JedisPoolUtil.getJedisObject();
            String jobId = String.format(OPENPROJECT_JOB_RECORD, value.getId());
            try {
                if (value.getTouser() == null || jedisObject.exists(jobId)) {
                    log.warn(value.getId() + "不能重复消费或找不到发送消息对象");
                    return;
                }
                log.info(value.toString());
                String userid = jedisObject.hget(OPENPROJECT_DINGDING_YS, value.getTouser());
                if (userid == null || "".equals(userid)) {
                    log.warn("不存在"+value.getTouser()+"相应的映射表");
                    return;
                }
                value.setTouserId(userid);
                value.setMsg(String.format(WORD_MESSAGE, value.getTouser(), value.getProjectName(), value.getWbsSubject(), value.getFromuser()));
                DingdingUtil.sendWorkmessage(value);
                jedisObject.set(jobId, JSON.toJSONString(value));
                jedisObject.expire(jobId, 15 * 60);//缓存十五分钟
            } finally {
                JedisPoolUtil.close(jedisObject);//回收redis对象
            }
        });
    }

    @Override
    public synchronized void open(Configuration parameters) {
        log.info("执行DingDingMessageSink的open方法");
    }

    private DingDingMessageSink() {
    }

    private static class Sigleton {
        private static final DingDingMessageSink dingDingMessageSink = new DingDingMessageSink();
    }

    public static DingDingMessageSink getInstance() {
        return Sigleton.dingDingMessageSink;
    }

}
